package active_device
import java.text.MessageFormat.format
import java.text.SimpleDateFormat
import java.util.{Calendar, TimeZone}
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.{SparkConf, SparkContext}
import org.json4s.jackson.JsonMethods._
import org.json4s._
import org.json4s.JsonDSL._
import org.json4s.jackson.JsonMethods._
import  org.json4s.JsonAST._
import com.typesafe.config._
import util.Util._
import  org.apache.spark.sql.types._
object daily_active_device_list {

  def main(args: Array[String]) {
    // Validate args
    if (args.length == 0) {
      println("Usage: [2015-05-25|date]")
       sys.exit(1)
    }
    implicit lazy val formats = org.json4s.DefaultFormats
    //  val date_str="2015-05-26"
    val date_str=args(0).toString
    val date_sdf = new SimpleDateFormat("yyyy-MM-dd")
    val daily_date = date_sdf.parse(date_str)
    val calender = Calendar.getInstance(TimeZone.getTimeZone("GMT+8"))
    calender.setTime(daily_date)
    val (year,month,day) =(calender.get(Calendar.YEAR),calender.get(Calendar.MONTH)+1,calender.get(Calendar.DAY_OF_MONTH))


    val daily_ac_dev_sql= format(getConfig("active_device.day_list"),year.toString,month.toString,day.toString,date_str.toString)

    // hdfs 激活设备存放目录
    val filePath=getConfig("active_device.hdfsdir")+"e_device_living_list_daily/"+date_str

    val sparkConf=new SparkConf().setAppName("day_active_device_list")
    val sc= new SparkContext(sparkConf)
    val hdfs=FileSystem.get(new Configuration())
    if(hdfs.exists(new Path(filePath)))hdfs.delete(new Path(filePath),true)

    val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc)
    sqlContext.sql(daily_ac_dev_sql).toJSON.coalesce(1, shuffle = true).saveAsTextFile(filePath)

   // df: org.apache.spark.sql.DataFrame = [ts: bigint, product_key: string, gid: string, mac: string, ip: string, country: string, region: string, city: string]
  /* df.map(line=>line.mkString(",")).map(line=>line.split(",")).map(line=>
      ("report_date"->date_str) ~ ("ts"->replaceNull(line(0)))~ ("product_key"->replaceNull(line(1)))~ ("group_id"->replaceNull(line(2)))~ ("mac"->replaceNull(line(3)))~ ("ip"->replaceNull(line(4)))~ ("country"->replaceNull(line(5)))~ ("region"->replaceNull(line(6)))~ ("city"->replaceNull(line(7)))
    ).map(line=>compact(render(line))).coalesce(1, shuffle = true).saveAsTextFile(filePath)
*/
    //pretty  ,compact
  }


}
